package core;

import java.io.IOException;
import java.nio.channels.*;
import java.nio.channels.spi.SelectorProvider;
import java.util.Iterator;
import java.util.Queue;
import java.util.Set;
import java.util.concurrent.LinkedBlockingQueue;

public class EventLoop implements Runnable{

    private final Thread thread;
    /**
     * 复用器，当前线程只处理这个复用器上的channel
     */
    public Selector selector;
    /**
     * 待处理的注册任务
     */
    private final Queue<Runnable> queue = new LinkedBlockingQueue<>();

    public EventLoop() throws IOException {
        this.selector = SelectorProvider.provider().openSelector();
        this.thread = new Thread(this);
        thread.start();
    }

    public void register(SocketChannel socketChannel, int keyOps,int timeout,Handler... handlers){
        // 将注册新的socketChannel到当前selector封装成一个任务
        queue.add(()->{
            try {
                MyChannel myChannel = new MyChannel(socketChannel, this);
                myChannel.setTimeout(timeout);
                for (Handler handle:handlers) {
                    myChannel.addLast(handle);
                }
                SelectionKey key = socketChannel.register(selector, keyOps);
                key.attach(myChannel);
            } catch (Exception e){
                e.printStackTrace();
            }
        });
        // 唤醒阻塞等待的selector线程
        selector.wakeup();
    }

    private void rebuildSelector() {
        final Selector oldSelector = selector;
        final Selector newSelector;

        if (oldSelector == null) {
            return;
        }

        try {
            newSelector=SelectorProvider.provider().openSelector();
        } catch (Exception e) {
            System.out.println("Failed to create a new Selector.");
            return;
        }

        // Register all channels to the new Selector.
        for (SelectionKey key: oldSelector.keys()) {
            Object a = key.attachment();
            try {
                if (!key.isValid() || key.channel().keyFor(newSelector) != null) {
                    continue;
                }

                int interestOps = key.interestOps();
                key.cancel();
                SelectionKey newKey = key.channel().register(newSelector, interestOps, a);
                newKey.attach(a);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }

        selector = newSelector;
    }

    /**
     * 循环地处理 注册事件、读写事件
     */
    @Override
    public void run() {
        while (!thread.isInterrupted()){
            int select = 0;//.selectNow();//
            try {
                select = selector.select(1000);
            } catch (IOException e) {
                System.out.println(Thread.currentThread() +"selector IO 异常");
                //需要重建selector
                rebuildSelector();
            }

            // 处理注册到当前selector的事件
            if (select == 0){
                Runnable task;
                while ((task = queue.poll()) != null){
                    task.run();
                }

                // 处理超时
                for (SelectionKey key:selector.keys()) {
                    MyChannel myChannel = (MyChannel) key.attachment();
                    myChannel.checkTimeOut(1000);
                }

                continue;
            }

            // 处理读写事件
            System.out.println(Thread.currentThread() +"服务器收到读写事件,select:" + select);
            processReadWrite();
        }
    }

    private void processReadWrite(){
        // 遍历事件进行处理
        Set<SelectionKey> selectionKeys = selector.selectedKeys();
        Iterator<SelectionKey> iterator = selectionKeys.iterator();
        while(iterator.hasNext()){
            MyChannel myChannel=null;
            try {
                SelectionKey key = iterator.next();
                myChannel = (MyChannel) key.attachment();
                if (key.isReadable()) {
                    // 将数据读进buffer
                    System.out.println(Thread.currentThread() + "开始读事件");
                    myChannel.doRead(key);
                }
                if (key.isWritable()) {
                    System.out.println(Thread.currentThread() + "开始写事件");
                    myChannel.doWrite(key);
                }
            }catch (CancelledKeyException e) {
                // 通道已关闭
                System.out.println("key已取消"+(e.getMessage()==null?"":e.getMessage()));
                if (myChannel!=null&& myChannel.isOpen()) myChannel.close();
            }
            iterator.remove();
        }
    }
}
